package net.bw.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bw.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String, String, String> {

    // 1. 声明状态
    private transient ValueState<String> firstDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 2.初始化状态
        firstDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("first-date", String.class)
        );
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // 3.获取状态值
        String firstDate = firstDateState.value();

        // 4.解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 获取 is_new 字段值
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        // 获取时间戳
        Long tsValue = jsonObject.getLong("ts");

        // 5.判断是否是新用户
        if ("1".equals(isNewValue)) {
            if (null == firstDate) {
                // 更新状态值
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            } else {
                // 判断当前时间是否和状态值相等，如果不相等，则修改is_new字段为0
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))) {
                    jsonObject.getJSONObject("common").put("is_new", "0");
                }
            }
        } else {
            //当用户提前登录，在处理时，用户第一次访问，firstDate为null，则更新firstDate为昨天的时间
            if (null == firstDate) {
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 86400000L, DateTimeUtil.DATE_FORMAT));
            }
        }

        out.collect(jsonObject.toJSONString());
    }
}
